// -*-mode:c++; coding:utf-8-*-

#include <cstring>

#include "sstable.hpp"

#include "sstable_scanner.hpp"

namespace xbase 
{

	sstable::pointer sstable::load(const std::string &real_path,
				       const std::string &sstable_path,
				       const std::string &deleted_sstable_path,
				       const std::string &start_row,
				       const std::string &end_row,
				       uint64_t sstable_sequence_number,
				       sstable_type type)
	{
		sstable::pointer _sstable(sstable::factory_type::construct(real_path,
									  sstable_path,
									  "", // compacting_sstable_path
									  deleted_sstable_path,
									  start_row,
									  end_row,
									  sstable_sequence_number,
									  0, // block size, not needed for loading
									  compressor::NONE, // also not needed
									  type));
		_sstable->init_for_loading();

		bool ok = _sstable->load_trailer();
		if (!ok)
		{
			return sstable::pointer();
		}
		
		ok = _sstable->load_index();
		if (!ok)
		{
			return sstable::pointer();
		}

		return _sstable;
	}

	// 只在加载时调用，不必对文件进行加锁
	bool sstable::load_trailer()
	{
		if(! check_file_for_read())
		{
			return false;
		}

		const fs::offset_t trailer_size = sstable_trailer::ST_LENGTH;
		const fs::offset_t trailer_pos = - trailer_size;
		const fs::offset_t seek_ret = fs::seek(m_file, trailer_pos, fs::ST_SEEK_END);
		if(seek_ret < 0)
		{
			LOG(ERROR, "sstable::load_trailer seek for trailer failed, pos = "
			    << trailer_pos << ", fs::errno = "
			    << fs::get_errno() << ", sstable real path = "
			    << m_real_path);
			return false;
		}

		const boost::asio::mutable_buffer buffer = m_file_buffer.prepare(sstable_trailer::ST_LENGTH);
		const fs::ssize_t ret = fs::readn(m_file, buffer);
		if(ret != static_cast<fs::ssize_t>(sstable_trailer::ST_LENGTH))
		{
			LOG(ERROR, "sstable::load_trailer read failed, ret = "
			    << ret << ", size to read = "
			    << sstable_trailer::ST_LENGTH
			    << "fs::errno = " << fs::get_errno()
			    << ", sstable real path = " << m_real_path);
			return false;
		}

		iarchive_type in = get_iarchive(buffer);
		const bool ok_decode = m_trailer.load(in);
		const bool ok = (ok_decode && m_trailer.valid());
		LOG_IF((! ok),
		       ERROR,
		       "sstable::load_trailer trailer decode or check failed, trailer = "
		       << m_trailer.to_string() << ", used buffer size = "
		       << in.used_buffer_size() << ", sstable real path = "
		       << m_real_path);
		
		return ok;
	}

	// 只在加载时调用，不必对文件进行加锁
	bool sstable::load_index()
	{
		if(! check_file_for_read())
		{
			return false;
		}

		// 首先读取索引的全部数据到内存
		const uint64_t index_data_begin = m_trailer.m_index_key_offset;
		const uint64_t index_data_end = m_trailer.m_file_size - sstable_trailer::ST_LENGTH;
		const uint64_t index_size = index_data_end - index_data_begin;

		const boost::asio::mutable_buffer buffer = m_file_buffer.prepare(index_size);
		const bool ok = load_data(buffer,
					  static_cast<fs::offset_t>(index_data_begin));
		if(! ok)
		{
			LOG(ERROR, "sstable::load_index load index data failed");
			return false;
		}
		iarchive_type in = get_iarchive(buffer);

		// 从内存中加载索引的Key部分
		sstable_block_header key_header;
		if(! key_header.load(in))
		{
			LOG(ERROR, "sstable::load_index decode index key block header failed, key header = "
			    << key_header.to_string() << ", sstable real path = " << m_real_path);
			return false;
		}
		if(! key_header.valid(sstable_block_header::BT_INDEX_KEY_BLOCK))
		{
			LOG(ERROR, "sstable::load_index index key header validation failed, key header = "
			    << key_header.to_string() << ", sstable real path = " << m_real_path);
			return false;
		}
		// 解压Key数据到目标内存
		const char * const key_data = in.left_buffer();
		// 跳过数据区；必须在解压前seek，也可同时判断剩下的数据大于header中的zsize
		if(! in.seek(key_header.m_data_zsize, SEEK_CUR))
		{
			LOG(ERROR, "sstable::load_index iarchive seek error, "
			    "maybe index key header is bad, key header = "
			    << key_header.to_string() << ", left data size = "
			    << in.left_buffer_size() << ", sstable real path = "
			    << m_real_path);
			return false;
		}
		const compressor::compressor_code code_key
			= static_cast<compressor::compressor_code>(key_header.m_compression_type);
		m_index_key_data.clear();
		const ssize_t key_dec_ret =
			compressor::decompress(code_key, m_index_key_data,
					       boost::asio::buffer(key_data, key_header.m_data_zsize));
		if((key_dec_ret < 0)
		   || (m_index_key_data.size() != key_header.m_data_size))
		{
			LOG(ERROR, "sstable::load_index decompressing index key data failed, dec_ret = "
			    << key_dec_ret << ", key header = "
			    << key_header.to_string() << " , sstable real path = " << m_real_path);
			return false;
		}

		// 加载Offset区数据
		sstable_block_header offset_header;
		if(! offset_header.load(in))
		{
			LOG(ERROR, "sstable::load_index decode index offset block "
			    "header failed, offset header = "
			    << offset_header.to_string() << ", sstable real path = "
			    << m_real_path);
			return false;
		}
		if(! offset_header.valid(sstable_block_header::BT_INDEX_OFFSET_BLOCK))
		{
			LOG(ERROR, "sstable::load_index index offset header "
			    "validation failed, offset header = "
			    << offset_header.to_string() << ", sstable real path = "
			    << m_real_path);
			return false;
		}
		// 判断剩下的数据是否正好和offset数据大小一致，否则就是出问题了
		if(in.left_buffer_size() != offset_header.m_data_zsize)
		{
			LOG(ERROR, "sstable::load_index index offset size mismatch, "
			    "maybe index offset header is bad, offset header = "
			    << offset_header.to_string() << ", real data zsize = "
			    << in.left_buffer_size() << ", sstable real path = "
			    << m_real_path);
			return false;
		}
		// 解压Offset区数据
		boost::asio::const_buffer offset_original_data;
		simple_buffer offset_buffer; // 函数执行完就会释放掉
		const compressor::compressor_code code_offset
			= static_cast<compressor::compressor_code>(offset_header.m_compression_type);
		if(code_offset == compressor::NONE)
		{
			if(offset_header.m_data_zsize != offset_header.m_data_size)
			{
				LOG(ERROR, "sstable::load_index index offset header's size "
				    << offset_header.m_data_size
				    << " not match to data zsize "
				    << offset_header.m_data_zsize
				    << " with compressor::NONE,"
				    " sstable real path = " << m_real_path);
				return false;
			}
			offset_original_data = boost::asio::buffer(in.left_buffer(),
								   offset_header.m_data_zsize);
		}
		else
		{
			const ssize_t offset_dec_ret =
				compressor::decompress(code_offset, offset_buffer,
						       boost::asio::buffer(in.left_buffer(),
									   offset_header.m_data_zsize));
			if((offset_dec_ret < 0)
			   || (offset_buffer.size() != offset_header.m_data_size))
			{
				LOG(ERROR, "sstable::load_index decompressing index offset "
				    "data failed, dec_ret = "
				    << offset_dec_ret << ", offset header = "
				    << offset_header.to_string()
				    << " sstable real path = " << m_real_path);
				return false;
			}
			offset_original_data = offset_buffer.data();
		}

		const boost::asio::const_buffer key_original_data = m_index_key_data.data();
		if(! build_index(key_original_data, offset_original_data))
		{
			LOG(ERROR, "sstable::load_index build index failed, sstable path = "
			    << m_sstable_path);
			return false;
		}
		return true;
	}
	
	sstable::pointer sstable::create(const std::string &sstable_path,
					 const std::string &compacting_sstable_path,
					 const std::string &deleted_sstable_path,
					 const std::string &start_row,
					 const std::string &end_row,
					 uint64_t sstable_sequence_number,
					 uint64_t block_size,
					 compressor::compressor_code compressor)
	{
		// create compact_file
		fs::file_t file = fs::create(compacting_sstable_path);
		if(file == fs::BAD_FILE)
		{
			LOG(ERROR, "sstable::create create compact file "
			    << compacting_sstable_path
			    << " failed, fs::errno = "
			    << fs::get_errno());
			return sstable::pointer();
		}

		sstable::pointer _sstable(sstable::factory_type::construct(compacting_sstable_path, // real path currently
									  sstable_path,
									  compacting_sstable_path,
									  deleted_sstable_path,
									  start_row,
									  end_row,
									  sstable_sequence_number,
									  block_size,
									  compressor,
									  ST_NORMAL_SSTABLE));
		_sstable->m_file = file;
		_sstable->init_for_creating();
		return _sstable;
	}

	bool sstable::append(const key_wrapper *key, const value_wrapper *value)
	{
		if(m_write_status == NULL)
		{
			LOG(ERROR, "sstable::append non-writable, sstable real path = "
			    << m_real_path);
			return false;
		}

		if(m_write_status->m_tmp_data_block_buffer.size() >= m_write_status->m_uncompressed_block_size)
		{
			const bool ok = append_data_block(m_write_status->m_tmp_data_block_buffer.data());
			if(! ok)
			{
				LOG(ERROR, "sstable::append append data block failed, trailer = "
				    << m_trailer.to_string() << ", sstable real path = "
				    << m_real_path);
				return false;
			}

			// 先添加索引，再clear数据缓存
			add_index(m_write_status->m_last_key, m_trailer.m_file_size); // with end offset

			// update compression rate
			m_write_status->m_uncompressed_block_size = 
				((m_trailer.m_block_size * m_trailer.m_data_size)
				 / m_trailer.m_data_zsize);

			// clear & reset data block buffer
			m_write_status->m_tmp_data_block_buffer.clear();
			// 保证块空间足够大，避免拷贝，现在重新分配内存
			// 代价是最小的，因为没有数据需要拷贝。多增加一
			// 个原始块大小，可以有效的避免添加最后一条记录
			// 导致内存不足
			m_write_status->m_tmp_data_block_buffer.prepare(m_write_status->m_uncompressed_block_size
									+ m_trailer.m_block_size);
		}

		const std::size_t key_bytes = key->bytes();
		const std::size_t value_bytes = value->bytes();
		const std::size_t bytes = key_bytes + value_bytes;
		boost::asio::mutable_buffer buffer = m_write_status->m_tmp_data_block_buffer.prepare(bytes);
		std::memcpy(boost::asio::buffer_cast<char*>(buffer), key, key_bytes);
		m_write_status->m_last_key = wrapper_cast<const key_wrapper*>(buffer);
		buffer = buffer + key_bytes;
		std::memcpy(boost::asio::buffer_cast<char*>(buffer), value, value_bytes);
		m_write_status->m_tmp_data_block_buffer.commit(bytes);

		++m_trailer.m_data_record_number;
		
		return true;
	}

	bool sstable::append_eof()
	{
		if(m_write_status == NULL)
		{
			LOG(ERROR, "sstable::append_eof non-writable sstable, sstable real path = "
			    << m_real_path);
			return false;
		}

		// write last block
		if(m_write_status->m_tmp_data_block_buffer.size() > 0)
		{
			const bool ok = append_last_data_block(m_write_status->m_tmp_data_block_buffer.data());
			if(! ok)
			{
				LOG(ERROR, "sstable::append_eof append last data block failed, trailer = "
				    << m_trailer.to_string() << ", sstable real path = "
				    << m_real_path);
				return false;
			}
			// 先添加索引，再clear数据缓存
			add_index(m_write_status->m_last_key, m_trailer.m_file_size); // with end offset
			m_write_status->m_tmp_data_block_buffer.clear();
		}

		// write index key data
		m_index_key_data.clear(); // swap tmp key data to key data buffer
		m_index_key_data.swap(m_write_status->m_tmp_index_key_buffer);
		const bool ok_key = append_index_key_block(m_index_key_data.data());
		if(! ok_key)
		{
			LOG(ERROR, "sstable::append_eof write index key block failed, trailer = "
			    << m_trailer.to_string() << ", sstable real path = "
			    << m_real_path);
			return false;
		}

		// write index offset data
		const bool ok_offset
			= append_index_offset_block(m_write_status->m_tmp_index_offset_buffer.data());
		if(! ok_offset)
		{
			LOG(ERROR, "sstable::append_eof write index offset block failed, trailer = "
			    << m_trailer.to_string() << ", sstable real path = "
			    << m_real_path);
			return false;
		}

		// write tailer
		const bool ok_trailer = append_trailer();
		if(! ok_trailer)
		{
			LOG(ERROR, "sstable::append_eof write trailer failed, trailer = "
			    << m_trailer.to_string() << ", sstable real path = "
			    << m_real_path);
			return false;
		}

		// close file to flush
		const bool ok_close = fs::close(m_file);
		if(! ok_close)
		{
			LOG(WARN, "sstable::append_eof close file failed, fs::errno = "
			    << fs::get_errno() << ", sstable real path = "
			    << m_real_path << ", ignore it");
		}
		m_file = fs::BAD_FILE;

		// build index
		const bool ok_build = build_index(m_index_key_data.data(),
						  m_write_status->m_tmp_index_offset_buffer.data());
		if(! ok_build)
		{
			LOG(ERROR, "sstable::append_eof build index failed, index key size = "
			    << m_index_key_data.size() << ", index offset size = "
			    << m_write_status->m_tmp_index_offset_buffer.size()
			    << ", trailer = " << m_trailer.to_string()
			    << ", sstable real path = " << m_real_path);
			return false;
		}

		// clear buffer
		m_write_status->clear();
		m_write_status = NULL;
		m_file_buffer.clear();

		// rename compact sstable filename
		const bool ok_rename = fs::rename(m_real_path, m_sstable_path);
		if(! ok_rename)
		{
			LOG(ERROR, "sstable::append_eof rename file("
			    << m_real_path << " -> " << m_sstable_path << ") failed, fs::errno = "
			    << fs::get_errno());
			return false;
		}
		m_real_path = m_sstable_path;
		
		return true;
	}

	bool sstable::append_block(const boost::asio::const_buffer &data,
				   sstable_block_header::block_type block_type)
	{
		sstable_block_header header(block_type);
		header.m_data_size = boost::asio::buffer_size(data);
		header.m_compression_type = ((header.m_data_size >= MIN_COMPRESS_BYTES)
					     ? m_trailer.m_compression_type
					     : compressor::NONE); // 只有数据大于一定字节，才启用压缩

		// 先假定为不压缩，减少代码分支
		header.m_data_zsize = header.m_data_size;
		boost::asio::const_buffer zdata = data;
		
		const compressor::compressor_code code
			= static_cast<compressor::compressor_code>(header.m_compression_type);
		if(code != compressor::NONE)
		{
			m_file_buffer.clear(); // must clear it here
			const ssize_t ret
				= compressor::compress(code,
						       m_file_buffer,
						       data);
			if(ret < 0)
			{
				LOG(ERROR, "sstable::append_block compress block failed, error code = "
				    << ret << ", data size to compress = "
				    << boost::asio::buffer_size(data)
				    << ", block header = " << header.to_string());
				return false;
			}
			// reset zdata & zsize
			header.m_data_zsize = m_file_buffer.size();
			zdata = m_file_buffer.data();
		}
		
		const uint64_t block_size = header.m_data_zsize + sstable_block_header::SBH_LENGTH;

		// 当前数据和缓存数据加起来够多的话，就直接写文件，而不拷贝内存了
		const bool is_need_flush = ((block_size + m_write_status->m_tmp_data_buffer.size())
					    >= config::write_block_size);
		if(! is_need_flush) // save header & data to data buffer
		{
			const boost::asio::mutable_buffer block =
				m_write_status->m_tmp_data_buffer.prepare(block_size);
			oarchive_type out(get_oarchive(block));
			if(! header.save(out))
			{
				LOG(ERROR, "sstable::append_block encode header failed, header = "
				    << header.to_string() << ", block buffer size = "
				    << boost::asio::buffer_size(block)
				    << ", used buffer size = " << out.used_buffer_size());
				return false;
			}
			if(out.left_buffer_size() != header.m_data_zsize)
			{
				LOG(ERROR, "sstable::append_data_block left buffer size error, "
				    "expected data zsize = " << header.m_data_zsize
				    << ", actural left size = " << out.left_buffer_size());
				return false;
			}
			std::memcpy(out.left_buffer(),
				    boost::asio::buffer_cast<const char*>(zdata),
				    out.left_buffer_size());
			m_write_status->m_tmp_data_buffer.commit(block_size);
		}
		else		// 块已经足够大，直接写入文件；以避免不必要的内存拷贝。
				// 别忘记缓存中可能存在的数据
		{
			// 准备数据
			char header_buf[sstable_block_header::SBH_LENGTH];
			oarchive_type out(header_buf, sstable_block_header::SBH_LENGTH);
			if(! header.save(out))
			{
				LOG(ERROR, "sstable::append_block encode header failed, header = "
				    << header.to_string()
				    << ", used buffer size = " << out.used_buffer_size());
				return false;
			}
			const boost::array<boost::asio::const_buffer, 3> block =
				{{
						m_write_status->m_tmp_data_buffer.data(),
						boost::asio::buffer(header_buf),
						zdata
					}};
			const uint64_t size = (boost::asio::buffer_size(zdata)
					       + sstable_block_header::SBH_LENGTH
					       + m_write_status->m_tmp_data_buffer.size());

			if(! check_file_for_write())
			{
				return false;
			}
			const fs::ssize_t ret = fs::writen(m_file, block);
			m_write_status->m_tmp_data_buffer.clear();
			if(ret != static_cast<fs::ssize_t>(size))
			{
				LOG(ERROR, "sstable::append_block direct-write data failed, ret = "
				    << ret << ", fs::errno = " << fs::get_errno()
				    << ", data size to be writen = "
				    << size
				    << ", sstable real path = " << m_real_path);
				return false;
			}
		}

		// update trailer info
		if(block_type == sstable_block_header::BT_DATA_BLOCK)
		{		// 只有是数据块时，才更新的域
			m_trailer.m_data_size += header.m_data_size;
			m_trailer.m_data_zsize += header.m_data_zsize;
			m_trailer.m_data_block_number += 1;
		}
		m_trailer.m_file_size += block_size;
		
		return true;
	}

	bool sstable::append_trailer()
	{
		m_trailer.m_timestamp = cur_micro_seconds();
		m_trailer.m_file_size += sstable_trailer::ST_LENGTH;
		
		// 生成数据
		oarchive_type out(
			get_oarchive(
				m_write_status->m_tmp_data_buffer.prepare(sstable_trailer::ST_LENGTH)
				)
			);
		if(! m_trailer.save(out))
		{
			LOG(ERROR, "sstable::append_trailer encode trailer failed, trailer = "
			    << m_trailer.to_string() << ", used buffer size = "
			    << out.used_buffer_size());
			return false;
		}
		m_write_status->m_tmp_data_buffer.commit(sstable_trailer::ST_LENGTH);
		
		// 写文件
		if(! check_file_for_write())
		{
			return false;
		}
		const boost::asio::const_buffer data_to_be_wroten
			= m_write_status->m_tmp_data_buffer.data();
		const fs::ssize_t ret = fs::writen(m_file, data_to_be_wroten);
		m_write_status->m_tmp_data_buffer.clear();
		if(ret != static_cast<fs::ssize_t>(boost::asio::buffer_size(data_to_be_wroten)))
		{
			LOG(ERROR, "sstable::append_trailer write data failed, ret = "
			    << ret << ", fs::errno = " << fs::get_errno()
			    << ", data size to be writen = "
			    << boost::asio::buffer_size(data_to_be_wroten)
			    << ", sstable real path = " << m_real_path);
			return false;
		}
		
		return true;
	}

	// build index in index key buffer
	bool sstable::build_index(boost::asio::const_buffer key,
				  boost::asio::const_buffer offset)
	{
		typedef integer_wrapper<fs::offset_t, kb::network_byteorder> offset_wrapper;

		m_index.clear();
		m_index.reserve(m_trailer.m_data_block_number);
		
		while(boost::asio::buffer_size(key) != 0)
		{
			const key_wrapper * const _key = wrapper_cast<const key_wrapper*>(key);
			const offset_wrapper * const _offset = wrapper_cast<const offset_wrapper*>(offset);
				
			m_index.push_back(index_entry(_key, _offset->get()));
				
			key = key + _key->bytes();
			offset = offset + _offset->bytes();
		}
		if(boost::asio::buffer_size(offset) != 0)
		{
			LOG(ERROR, "sstable::build_index index offset data mismatch, left "
			    << boost::asio::buffer_size(offset) << " offset data, sstable real path = "
			    << m_real_path);
			return false;
		}

		if(m_trailer.m_data_block_number != m_index.size()) // 块数即是索引条数
		{
			LOG(ERROR, "sstable::build_index index entry number mismatch, "
			    "trailer.m_data_block_number = "
			    << m_trailer.m_data_block_number << ", index.size = " << m_index.size()
			    << ", sstable real path = " << m_real_path);
			return false;
		}
		// 合适的时机，这种读、写两种方式的情况都处理到了
		return update_split_row_and_filesize();
	}

	void sstable::add_index(const key_wrapper *key, const fs::offset_t &offset)
	{
		const size_t key_bytes = key->bytes();
		boost::asio::mutable_buffer tmp = m_write_status->m_tmp_index_key_buffer.prepare(key_bytes);
		std::memcpy(boost::asio::buffer_cast<char*>(tmp), key, key_bytes);
		m_write_status->m_tmp_index_key_buffer.commit(key_bytes);

		// convert offset data
		typedef integer_wrapper<fs::offset_t, kb::network_byteorder> offset_wrapper;
		offset_wrapper off;
		off.set(offset);
		const size_t off_bytes = off.bytes();
		tmp = m_write_status->m_tmp_index_offset_buffer.prepare(off_bytes);
		std::memcpy(boost::asio::buffer_cast<char*>(tmp), &off, off_bytes);
		m_write_status->m_tmp_index_offset_buffer.commit(off_bytes);
	}

	// load block and commit to cache manager and check out
	sstable_block_cache sstable::get_block(std::size_t index) const
	{
		if(! check_index(index))
		{
			LOG(ERROR, "sstable::get_block index out-of-range, index = "
			    << index << ", index range = ["
			    << m_index_begin << ", "
			    << m_index_end << "), index size = "
			    << m_index.size() << ", sstable type = "
			    << sstable_type_name(m_sstable_type)
			    << ", sstable path = " << m_sstable_path
			    << ", sstable real path = " << m_real_path);
			return sstable_block_cache();
		}
		
		// check if it's cached
		{
			sstable_block_cache block(m_id, index);
			if(! (! block))
			{
				return block;
			}
		}

		// get file lock, guard for other read file op
		boost::mutex::scoped_lock file_guard(m_file_lock);
		
		// double check, in case it has been loaded by other threads
		{
			sstable_block_cache retry(m_id, index);
			if(! (! retry))
			{
				return retry;
			}
		}

		if(! check_file_for_read())
		{
			return sstable_block_cache();
		}

		// 确认要读后面的多少个块。要考虑连续、后面有几个块没有
		// 缓存、一次读取的最大数据大小、缓存区中的剩余内存。如
		// 果没有启用缓存，则只读当前块。
		//
		const std::size_t start_index = index;
		const fs::offset_t start_offset = ((start_index == 0)
						   ? 0
						   : m_index[start_index - 1].m_offset);
		std::size_t end_index = start_index + 1;
		uint64_t read_size = m_index[end_index - 1].m_offset - start_offset;

		// TODO:注意这里算的SSTable块大小是压缩后的大小，而不是
		// 原数据大小；缓冲中存放的块是解压后的；所以这里读入的
		// 块数据不能太多。也许再加上块数的限制会好些；
		//
		// 最好的解决方式是将块的信息（部分或全部块header）放到
		// 索引中，这样就可以精确的知道一个块的原数据有多大了。

		// 读取的总数据大小比config::read_block_size大一点没关
		// 系，但一定不能比可用内存大
		while((end_index < m_index_end)
		      && (read_size < config::read_block_size)
		      && (! (sstable_block_cache::is_cached(m_id, end_index)))) // 如果下一个块没有被缓存
		{
			// 预算加上下个块的总数据大小
			const uint64_t total = (m_index[end_index].m_offset - start_offset);
			if(total > sstable_block_cache_manager::instance().available_memory())
			{
				break; // 停止
			}

			++ end_index; // 包含进来
			read_size = total; // 更新read_size
		}
		// 这样就确定了，要从start_offset开始，读取到read_size
		// 大小的数据，包含了[start_index, end_index)这些块

		// 加载数据
		const boost::asio::mutable_buffer buffer = m_file_buffer.prepare(read_size);
		const bool load_ok = load_data(buffer, start_offset);
		if(! load_ok)
		{
			LOG(ERROR, "sstable::get_block load data blocks failed, range = [" << start_index
			    << ", " << end_index << "), start offset = " << start_offset
			    << ", size = " << read_size << ", sstable real path = " << m_real_path);
			return sstable_block_cache();
		}

		// 为每个块分配一个缓存块，用来保存解析后待提交的数据
		const uint64_t n = end_index - start_index;
		sstable_block_buffer buffer_array[n];

		// 解析块
		iarchive_type in = get_iarchive(buffer);
		sstable_block_header header;
		uint64_t n_loaded = 0;
		for(uint64_t i = start_index; i < end_index; ++i, ++n_loaded)
		{
			if(! header.load(in))
			{
				LOG(ERROR, "sstable::get_block decode " << i
				    << "th block data failed, loading range = [" << start_index
				    << ", " << end_index << "), start offset = " << start_offset
				    << ", size = " << read_size << ", sstable real path = " << m_real_path);
				break;
			}
			if(! header.valid(sstable_block_header::BT_DATA_BLOCK))
			{
				LOG(ERROR, "sstable::get_block validation " << i
				    << "th block data header failed, header = "
				    << header.to_string() << " loading range = [" << start_index
				    << ", " << end_index << "), start offset = " << start_offset
				    << ", size = " << read_size << ", sstable real path = " << m_real_path);
				break;
			}

			const char * const data = in.left_buffer();
			// 跳过当前块的数据区，同时可以检查剩余数据是否足够
			if(! in.seek(header.m_data_zsize, SEEK_CUR))
			{
				LOG(ERROR, "sstable::get_block iarchive seek error at " << i
				    << "th block data, header = "
				    << header.to_string() << " loading range = [" << start_index
				    << ", " << end_index << "), start offset = " << start_offset
				    << ", size = " << read_size << ", sstable real path = " << m_real_path);
				break;
			}
			
			sstable_block_buffer &buffer = buffer_array[n_loaded];
			const compressor::compressor_code dec
				= static_cast<compressor::compressor_code>(header.m_compression_type);
			const ssize_t dec_ret
				= compressor::decompress(dec,
							 buffer,
							 boost::asio::buffer(data, header.m_data_zsize));
			if((dec_ret < 0)
			   || (buffer.size() != header.m_data_size))
			{
				LOG(ERROR, "sstable::get_block decompress " << i
				    << "th block failed, dec_ret = " << dec_ret << ", header = "
				    << header.to_string() << " loading range = [" << start_index
				    << ", " << end_index << "), start offset = " << start_offset
				    << ", size = " << read_size << ", sstable real path = " << m_real_path);
				break;
			}
			
			if(! buffer.build_index())
			{
				LOG(ERROR, "sstable::get_block build index on " << i
				    << "th block failed, loading range = [" << start_index
				    << ", " << end_index << "), start offset = " << start_offset
				    << ", size = " << read_size << ", sstable real path = " << m_real_path);
				break;
			}
		}
		assert(in.left_buffer_size() == 0); // 数据错误！
		if(n_loaded == 0) // 第一条就错误，则马上返回
		{
			LOG(ERROR, "sstable::get_block decode first block failed, "
			    "loading range = [" << start_index
			    << ", " << end_index << "), start offset = " << start_offset
			    << ", size = " << read_size << ", sstable path = " << m_sstable_path);
			return sstable_block_cache();
		}

		// 先提交预读的块（从后向前）
		// 
		// 从后向前顺序提交块缓存。这样即使缓存空间不足（现在或
		// 以后），也会使后面的块（先提交）先被淘汰，前面的块
		// （后提交）后被淘汰，从而保证了缓存中块的连续性，这对
		// 顺序读和块的预读都有好处。
		while(n_loaded > 1)
		{
			-- n_loaded; // --后，可用作下标

			const uint64_t no = (start_index + n_loaded);
			const bool ok =
				buffer_array[n_loaded].commit(m_id,
							      no);
			// 若失败，只记录日志，不返回错误，继续提交前面的块
			if(! ok)
			{
				LOG(WARN, "sstable::get_block commit " << no
				    << "th block failed, loading range = [" << start_index
				    << ", " << end_index << "), start offset = " << start_offset
				    << ", size = " << read_size << ", available cache memory = "
				    << sstable_block_cache_manager::instance().available_memory()
				    << ", cur commit size = "
				    << buffer_array[n_loaded].size()
				    //(m_index[no].m_offset - m_index[no - 1].m_offset)
				    << ", sstable path = " << m_sstable_path);
			}
		}

		// 最后提交用户要的那个块，即第一个块
		sstable_block_cache wanted_block =
			buffer_array[0].commit_and_checkout(m_id, start_index);
		if(! wanted_block) // failed
		{
			LOG(ERROR, "sstable::get_block commit and checkout " << start_index
			    << "th block failed, loading range = [" << start_index
			    << ", " << end_index << "), start offset = " << start_offset
			    << ", size = " << read_size << ", sstable path = " << m_sstable_path);
		}

		LOG(INFO, "sstable::get_block load data blocks ok, range = [" << start_index
		    << ", " << end_index << "), start offset = " << start_offset
		    << ", size = " << read_size << ", sstable path = " << m_sstable_path);
		
		// 仅在用户想要的块出现错误的时候，才返回错误；否则，只打印出错误日志，不返回错误。
		return wanted_block;
	}

	// 从索引中取中间一条索引的行作为分裂行
	bool sstable::update_split_row_and_filesize()
	{
		if(m_index.empty())
		{
			m_split_row = "";
			// 这时不要将文件大小设为0,而保留原来的值
			//m_trailer.m_file_size = 0;
			LOG(WARN, "sstable::update_split_row_and_filesize empty sstable, sstable path = "
			    << m_sstable_path);
			return true;
		}

		const std::size_t mid = m_index.size() >> 1;
		const std::string *row = NULL; // parent's split row
		if(is_ref_top_sstable())
		{
			row = &m_end_row;
		}
		else if(is_ref_bottom_sstable())
		{
			row = &m_start_row;
		}
		
		if(row == NULL)	// not ref sstable
		{
			m_split_row = (std::string)(*m_index[mid].m_key->row());
			m_index_begin = 0;
			m_index_end = m_index.size();
		}
		else
		{
			// 找split row原来的索引位置
			std::size_t i = mid;
			const int ret = m_index[i].m_key->row()->compare(*row);
			if(ret < 0) // 向后找
			{
				for(++i;
				    ((i < m_index.size())
				     && (m_index[i].m_key->row()->compare(*row) < 0));
				    ++i) { ; }
			}
			else if(ret > 0)
			{
				for(--i;
				    ((i > 0)
				     && (m_index[i].m_key->row()->compare(*row) >= 0)); // must be >= here
				    --i) { ; }
				++i;
				// 注意：上面一定要用>=；因为当分裂行的
				// 数据跨越多个块时，只有这样才能得到合
				// 理的结果
			}
			else
			{
				;
			}
			// i 就是索引分界线，该块被两个子SSTable共享

			// 既然已经找到了父子表的分裂行，那么就知道了自己文件的数据大小，在此更新！
			const uint64_t pos = m_index[i].m_offset;
			if(is_ref_top_sstable())
			{
				m_trailer.m_file_size = pos;
				if(i == 0)
				{
					m_split_row = "";
				}
				else
				{
					m_split_row = (std::string)(*m_index[i >> 1].m_key->row());
				}
				// update index range
				m_index_begin = 0;
				m_index_end = (i + 1);
			}
			else
			{
				if(m_trailer.m_file_size <= pos)
				{
					LOG(ERROR, "sstable::update_split_row_and_filesize"
					    " bad offset " << pos << " at " << i
					    << "th index, filesize = " << m_trailer.m_file_size
					    << ", sstable path = " << m_sstable_path);
					return false;
				}
				else
				{
					m_trailer.m_file_size = m_trailer.m_file_size - pos;
				}

				if(i >= (m_index.size() - 1))
				{
					m_split_row = "";
				}
				else
				{
					const std::size_t idx = ((m_index.size() - i) >> 1) + i;
					m_split_row = (std::string)(*m_index[idx].m_key->row());
				}
				// update index range
				m_index_begin = i;
				m_index_end = m_index.size();
			}
		}

		LOG(INFO, "sstable::update_split_row_and_filesize got split row = "
		    << dump_string_char(m_split_row) << "(hex:"
		    << dump_string_hex(m_split_row)
		    << "), file size = " << m_trailer.m_file_size
		    << ", index range = [" << m_index_begin
		    << ", " << m_index_end
		    << "), index size = " << m_index.size()
		    << ", sstable path = " << m_sstable_path);
		return true;
	}

	bool sstable::lower_bound(sstable_block_cache &got_block,
				  std::pair<size_t, size_t> &index_pair,
				  const key_wrapper *key)
	{
		index_type::const_iterator block_iter = std::lower_bound(m_index.begin(),
									 m_index.end(),
									 key,
									 index_entry_less_than());
		if(block_iter == m_index.end())
		{
			return false;
		}

		// lower_bound返回的这个块，说明该Key一定在这个块内，因
		// 为该块的最后一个Key已经大于等于给定Key了，若经实际查
		// 找又不在，说明是数据出错，应该报错！

		std::size_t block_pos = (block_iter - m_index.begin());
		sstable_block_cache block = get_block(block_pos); // 经确认后再传出去
		if(! block)
		{
			LOG(ERROR, "get block #" << block_pos << " failed");
			return false;
		}
		
		const block_index_type &block_index = block.block_index();
		block_index_type::const_iterator record_iter = std::lower_bound(block_index.begin(),
										block_index.end(),
										key,
										block_entry_less_than());
		if(record_iter == block_index.end()) // 这里检查又没有，说明数据有错误，报错，并返回错误！
		{
			LOG(FATAL, "sstable::lower_bound block data mismatch with "
			    "sstable index data!, sstable path = "
			    << m_sstable_path);
			return false;
		}

		const size_t record_pos = (record_iter - block_index.begin());

		got_block = block;
		index_pair.first = block_pos;
		index_pair.second = record_pos;

		return true;
	}

	// 将指定位置的记录取出（必要时加载块，以及更新索引为该记录真正
	// 的索引位置），同时返回记录所属的块
	//
	// 返回False，表示指定索引已经超过SSTable的所有记录
	// 返回True，则表示记录entry在block中，并且它的真实索引位置是index_pair
	//
	// 读完一条索引后，若需要下一条索引处的记录，则只简单的将块内索
	// 引加一即可。或调用advance_index()来完成
	bool sstable::get_entry(sstable_block_cache &block,
				block_entry &entry,
				std::pair<size_t, size_t> &index_pair)
	{
		// find the correct block
		do
		{
			if(index_pair.first >= m_index.size())
			{
				LOG(TRACE, "sstable::next_entry pos("
				    << index_pair.first << ", " << index_pair.second
				    << ") out of range, sstable path = "
				    << m_sstable_path);
				return false;
			}
		
			if(! block)
			{
				block = get_block(index_pair.first);
				if(! block)
				{
					LOG(ERROR, "get block #" << index_pair.first << " failed");
					return false;
				}
			}

			const block_index_type &block_index = block.block_index();
			if(index_pair.second >= block_index.size())
			{
				block.release(); // release cur block
				index_pair.first += 1; // and go next block
				index_pair.second = 0;
				continue;
			}
			else	// found
			{
				entry = block_index[index_pair.second];
				return true;
			}
		} while(true);
	}

	std::string sstable_trailer::to_string() const {
		std::ostringstream out;
		out << "{magic: 0x" << std::hex << m_magic << std::dec
		    << "; data_size: " << m_data_size
		    << "; data_zsize: " << m_data_zsize
		    << "; file_size: " << m_file_size
		    << "; compression_type: "
		    << compressor::compressor_name(static_cast<compressor::compressor_code>(m_compression_type))
		    << "; data_block_number: " << m_data_block_number
		    << "; data_record_number: " << m_data_record_number
		    << "; index_key_offset: " << m_index_key_offset
		    << "; index_offset_offset: " << m_index_offset_offset
		    << "; block_size: " << m_block_size
		    << "; timestamp: " << m_timestamp
		    << "}";
		return out.str();
	}

	std::string sstable_trailer::stat_info() const {
		std::ostringstream out;
		out.precision(4);
		out << "data compress rate: " << (static_cast<double>(m_data_zsize) / m_data_size) * 100 << "%";
		out << ", disk usage rate: " << (static_cast<double>(m_file_size) / m_data_size) * 100 << "%";
		return out.str();
	}

	std::string sstable_block_header::to_string() const {
		std::ostringstream out;
		out << "{magic:" << std::hex << m_magic << std::dec
		    << "; compression_type: "
		    << compressor::compressor_name(static_cast<compressor::compressor_code>(m_compression_type))
		    << "; data_size: " << m_data_size
		    << "; data_zsize: " << m_data_zsize
		    << "}";
		return out.str();
	}

	std::string sstable_block_header::stat_info() const {
		std::ostringstream out;
		out.precision(4);
		out << "data compress rate: "
		    << (static_cast<double>(m_data_zsize) / m_data_size) * 100
		    << "%";
		return out.str();
	}

	sstable::~sstable() {
		if(fs::BAD_FILE != m_file)
		{
			fs::close(m_file);
			m_file = fs::BAD_FILE;
		}

		if(m_write_status)
		{
			m_write_status->clear();
			m_write_status = NULL;
			// 若没设置删除标志，则应该将紧缩文件删除。因为
			// 检测到只紧缩到一半。
			if(! m_deleted)
			{
				const bool ok = fs::remove(m_compacting_sstable_path);
				LOG_IF_ELSE(ok,
					    DEBUG,
					    "sstable::~sstable broken compacting sstable "
					    << m_compacting_sstable_path << " deleted",
					    WARN,
					    "sstable::~sstable broken deleting compacting sstable "
					    << m_compacting_sstable_path << " failed");
				(void)ok;
			}
		}
		
		// SSTable销毁时，也将它的所有缓存都清空。此时
		// 要保证该SSTable的所有块缓存都没有正被使用
		//
		// 当前只有块缓存，以后可能有扫描缓存
		// 
#ifndef NDEBUG
		sstable_block_cache_manager::instance().clear_sstable_cache(m_id);
#endif	// NDEBUG

		if(m_deleted)
		{
			const bool ok = fs::remove(m_deleted_sstable_path);
			LOG_IF_ELSE((!ok),
				    ERROR,
				    "sstable::~sstable remove sstable file "
				    << m_deleted_sstable_path << " failed, fs::errno = "
				    << fs::get_errno(),
				    DEBUG,
				    "sstable::~sstable deleted sstable "
				    << m_deleted_sstable_path << " removed");
			(void)ok;
		}
	}

	bool sstable::set_deleted() {
		boost::mutex::scoped_lock file_guard(m_file_lock);

		if(m_deleted)	// 已经被删除了
		{
			return true;
		}
		
		if(m_file != fs::BAD_FILE)
		{
			fs::close(m_file);
			m_file = fs::BAD_FILE;
		}

		// 必须是real_path才能将处于compacting状态的文件删除
		const std::string *cur_path = &m_real_path;
		if(is_ref_sstable()) // 如果是引用SSTable，则只能动自己的文件
		{
			cur_path = &m_sstable_path;
		}
			
		const bool ok = fs::rename(*cur_path, m_deleted_sstable_path);
		if(! ok)
		{
			LOG(ERROR, "sstable::set_deleted rename file "
			    << *cur_path << " to " << m_deleted_sstable_path
			    << " failed, fs::errno = "
			    << fs::get_errno());
			return false;
		}
		// 重置real_path
		if(! is_ref_sstable()) // 如果是引用文件，则不能更改该路径
		{
			m_real_path = m_deleted_sstable_path;
		}
		
		m_deleted = true;
		return true;
	}

	kv_scanner::pointer sstable::get_scanner(const kv_scan_context::pointer &context)
	{
		return kv_scanner::pointer(new sstable_scanner(context, sstable::pointer(this)));
	}

	std::string sstable::to_string() const {
		std::ostringstream out;
		out << "{id:" << m_id
		    << "; sstable_path: " << m_sstable_path
		    << "; compacting_path: " << m_compacting_sstable_path
		    << "; deleted_path: " << m_deleted_sstable_path
		    << "; start_row: " << m_start_row
		    << "; end_row: " << m_end_row
		    << "; sequence_number: " << m_sequence_number
		    << "; sstable type: " << sstable_type_name(m_sstable_type)
		    << "; deleted: " << (m_deleted ? "true" : "false")
		    << "; real path: " << m_real_path
		    << "; trailer: " << m_trailer.to_string()
		    << "; split row: " << m_split_row
		    << "; index range: [" << m_index_begin
		    << ", " << m_index_end << ")"
		    << "; index_size: " << m_index.size()
		    << "}";
		return out.str();
	}

} // namespace xbase

